package com.young.study.mq.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class MyConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        //指定kafka服务器地址 如果是集群可以指定多个 但是就算只指定一个他也会去集群环境下寻找其他的节点地址
        properties.put("bootstrap.servers", "192.168.199.201:9092");
        // key 的反序列化器，和序列化器相对，不要弄混
        properties.put("key.deserializer", StringDeserializer.class.getName());
        // value 的反序列化器，和序列化器相对，不要弄换
        properties.put("value.deserializer", StringDeserializer.class.getName());
        // 消费者组id,同一个消息只能被同一个组的某一个消费者消费一次
        properties.setProperty("group.id","1111");
        // 此设置限制每次调用poll返回的消息数(默认 500)，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的
        properties.setProperty("max.poll.records","500");
        // 增两次poll的间隔（默认300000ms 即五分钟），可以为消费者提供更多的时间去处理返回的消息（调用poll(long)返回的消息，通常返回的消息都是一批）。缺点是此值越大将会延迟组重新平衡
        properties.setProperty("max.poll.interval.ms","300000");
        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        // 消费者订阅主题，可以订阅多个,也可以通过正则匹配的方式匹配多个主题，kafkaConsumer.subscribe("test.*");
        kafkaConsumer.subscribe(Collections.singletonList("my-topic"));
        // kafka 不同于其他消息队列，其他消息队列属于broker主动推送 消息，而kafka需要我们手动的去获取消息，
        // 如果要持续不断获取消息，这里可以使用while(true)方式实现
        ConsumerRecords<String, String> poll = kafkaConsumer.poll(500);
        for (ConsumerRecord<String, String> consumerRecord : poll) {
            System.out.println(consumerRecord);
        }
    }
}
